//package org.example;
//
//import org.apache.flink.api.common.functions.FlatMapFunction;
//import org.apache.flink.api.common.functions.ReduceFunction;
//import org.apache.flink.api.java.utils.ParameterTool;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.DataStreamSource;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.util.Collector;
//
///**
// * Hello world!
// *
// */
//public class App
//{
//    public static void main( String[] args )
//    {
////        final int port;
////        try {
////            ParameterTool params = ParameterTool.fromArgs(args);
////            port = params.getInt("port");
////        } catch (Exception e) {
////            System.err.println("No port specified. Please run 'SocketWindowWordCount --port <port>'");
////            return;
////        }
//
//        //获取执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        //通过socket获取输入数据
//        DataStreamSource<String> text = env.socketTextStream("localhost", 9000, "\n");
//
//        //聚合计算
//        DataStream<WordWithCount> windowCounts = text
//                .flatMap(new FlatMapFunction<String, WordWithCount>() {
//
//                    @Override
//                    public void flatMap(String value, Collector<WordWithCount> out) throws Exception {
//
//                        for (String word : value.split("\\s")) {
//                            out.collect(new WordWithCount(word, 1L));
//                        }
//
//                    }
//                })
//                .keyBy("word")
//                .timeWindow(Time.seconds(5), Time.seconds(1))
//                .reduce(new ReduceFunction<WordWithCount>() {
//                    @Override
//                    public WordWithCount reduce(WordWithCount a, WordWithCount b) throws Exception {
//                        return new WordWithCount(a.word, a.count + b.count);
//                    }
//                });
//
//        windowCounts.print().setParallelism(1);
//
//        try {
//            env.execute("socket Window WordCount");
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//
//    }
//
//    public static class WordWithCount{
//
//        public String word;
//        public Long count;
//
//        public WordWithCount(){}
//
//        public WordWithCount(String word, Long count) {
//            this.word = word;
//            this.count = count;
//        }
//
//        @Override
//        public String toString() {
//            return "WordWithCount{" +
//                    "word='" + word + '\'' +
//                    ", count=" + count +
//                    '}';
//        }
//    }
//}
